package com.hh.job;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.hh.entity.domain.Blog;
import com.hh.entity.dto.blog.PostEsDTO;
import com.hh.service.BlogService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author hanhui
 * @version 1.0
 * @description: TODO
 * @date 2025/10/7 15:26
 **/
@Component
@Slf4j
public class SyncBlog2EsJob {

    @Resource
    private BlogService blogService;
    @Resource
    private ElasticsearchRestTemplate elasticsearchRestTemplate;
    // 添加最后同步时间记录
    private volatile long lastSyncTime = 0L;

    @Scheduled(fixedRate = 10000)
    @Async
    @Transactional(rollbackFor = Exception.class)
    public void run() {
        log.info("定制任务开始，将数据库的文章信息同步至ES");
        SyncBlog2Es();
        log.info("定制任务结束，将数据库的文章信息同步至ES");
    }

    @Transactional(rollbackFor = Exception.class)
    public void SyncBlog2Es() {
        try {
            // 获取增量数据
            long currentTime = System.currentTimeMillis();
            List<Blog> blogList;

            if (lastSyncTime == 0) {
                // 首次全量同步
                QueryWrapper<Blog> queryWrapper = new QueryWrapper<>();
                queryWrapper.eq("isDelete", 0);
                blogList = blogService.list(queryWrapper);
                log.info("Performing full sync of blog posts");
            } else {
                // 增量同步，只获取更新时间大于上次同步时间的数据
                blogList = blogService.listByUpdateTimeAfter(lastSyncTime);
                log.info("Performing incremental sync of blog posts since {}", new Date(lastSyncTime));
            }

            if (blogList.isEmpty()) {
                log.info("No new blog posts to sync");
                lastSyncTime = currentTime;
                return;
            }

            List<PostEsDTO> postEsDTOList = blogList.stream().map(PostEsDTO::objToDto).collect(Collectors.toList());

            final int pageSize = 500;
            int total = postEsDTOList.size();
            int successCount = 0;

            log.info("SyncBlogToEs start, total {}", total);

            for (int i = 0; i < total; i += pageSize) {
                int end = Math.min(i + pageSize, total);
                log.info("sync from {} to {}", i, end);

                try {
                    elasticsearchRestTemplate.save(postEsDTOList.subList(i, end));
                    successCount += (end - i);
                } catch (Exception e) {
                    log.error("Failed to sync blog posts from {} to {}: {}", i, end, e.getMessage(), e);
                    // 记录失败的数据ID，可后续重试
                }
            }

            // 只有在同步成功后才更新最后同步时间
            lastSyncTime = currentTime;

            log.info("SyncBlogToEs end, total {}, success {}", total, successCount);
        } catch (Exception e) {
            log.error("Sync blog to ES failed: {}", e.getMessage(), e);
            throw e;
        }
    }

}